package com.toutiao.nssue

import scala.math.sqrt

import org.apache.log4j.Logger

import scala.collection.mutable.ArrayBuffer
import scala.collection.mutable.{HashMap => HMap}

import com.toutiao.issue.SimpleDate
import com.toutiao.issue.UpperDict

import com.toutiao.model.WAND

import com.toutiao.score.{Topic,SimpleFactory}
import com.toutiao.model.{Converter,TimeDict,TimeDB,TimeIndex}

import com.twitter.util.{SynchronizedLruMap => LruCache}

class NssueCollector (
					val minS: Double,
					val mIndex: TimeIndex,
					val mTime: TimeDict,
					val mDB: TimeDB,
					val mVU: UpperDict
    ) {
  
  private
  val _log = Logger.getLogger(this.getClass.getName)
  
  def get(n:String) = { // 使用Cache会更好些
    if (mTime.isIn(n)) {
      val t = mTime(n)
      val m = mDB(t)(n)
      m
    }
    else {
      //new Event("",Array(""),Map[String,Double](),SimpleDate.current,SimpleDate.current,0.0)
      new Topic
    }
  }
  
  private
  def norm(vs:Map[String,Double]) = {
    SimpleFactory.norm(vs)
  }

  private val MIN_SAME_TOKEN = 3
  private val MAX_SIZE_TOKEN = 35
  
  private
  def cos(m1:Map[String,Double], s1:Double, m2:Map[String,Double], s2:Double):Double = {
   SimpleFactory.cos(m1,s1,m2,s2)
  }

  private
  def merge(m1:Map[String,Double],n1:Int,m2:Map[String,Double],n2:Int) = {
    SimpleFactory.cut(SimpleFactory.merge(m1,n1,m2,n2),MAX_SIZE_TOKEN)
  }
  
  private
  def asTopic(x:Document) = {
    //new Event(x.name,Array(x.name),norm(x.vs),x.date,x.date,1.0)
    val e = new Topic
    e.i = x.name
    e.v = x.vs
    e.s = 1.0
    e.c = x.date
    e.u = x.date
    val ms = Array(x.name)
    e.m = Map("m"->ms)
    e
  }
  
  def getSimScore(m1:Topic,m2:Topic):Double= {
    cos(m1.v,m1.s,m2.v,m2.s)
  }
  
  def insert(name:String,i:Long,event:Topic) {
    val t = SimpleDate.format(event.c).split(" ")(0)
    mDB.insert(t, name, event)
    mIndex.insert(t, i, event.v.keySet.toArray)
    mTime.insert(name,t)
    event.v.foreach(vv => { // Update VU
      mVU.put(vv._1,vv._2.toFloat)
    })
  }
  
  def remove(t:String) = {
    mIndex.remove(t)
    mDB.remove(t)
    mTime.removeTime(t)
  }
  
  private
  def retrieve(vs:Map[String,Double],S:Double) = {
    val us = vs.map(vv => {
      val u = mVU.get(vv._1)
      (vv._1,vv._2*u)
    })
    val iis = mIndex.getIndex
    val es = iis.map(index => {
      new WAND(index,us,S)
    })
    
    val ns = es.par.flatMap(e => e.retrieve).seq
    ns
  }
  
  private
  val CACHE = new LruCache[String,Topic](65536*16)
  
  def getFeature(ns:Array[String]) = {    
    val nns = ns.filter(!CACHE.contains(_))
    val xxs = ns.filter(CACHE.contains(_)).map(n => {(n,CACHE(n))}) // Hit Cache
    val xs = nns.map(n => {
      val t = mTime(n)
      (n, mDB(t).getValue(n))
    })

    val xData = xs.par.map(x => {
      val n = x._1
      val v = Converter.parseData(x._2)
      (n,v)
    }).seq.toMap

    xData.foreach( nv => { // Update Cache
    	if (nv._2.m("m").size > 1)
    	  CACHE(nv._1) = nv._2
      }
    )

    xData++xxs
  }
  
  def getData(n:String) = {
    if (CACHE.contains(n)){
      CACHE(n)
    }
    else {
      val t = mTime(n)
      val x = mDB(t).getValue(n)
      Converter.parseData(x)
    }
  }
  
  def processOne(M1:HMap[String,Long],
                 M2:HMap[Long,String],
                 x:Document,
                 S:Double) = {
    
    var event = asTopic(x)
    var r = _processOne(M1,M2,event,S,true)
    var rs = List[String]()
    if (r == event.i){
      rs = rs :+ r
    }
    else {
      while ( r != "" && r != event.i){
        rs = rs :+ r
        event = getData(r)
        r = _processOne(M1,M2,event,S,false)
        if (r == event.i){
          _log.error(x.name+" "+event.i)
        }
      }
    }

    rs
  }
  
  def _processOne(
		  	M1:HMap[String,Long],
		  	M2:HMap[Long,String],
		  	event: Topic,
		  	S: Double,
		  	isNew:Boolean = true
      ):String = {

    val ns = retrieve(event.v,minS).
    			filter(M2.contains(_)).
    			map(M2(_)).toArray

    val xData = getFeature(ns)
    
    val rs = ns.filter(_ != event.i).par.map(n => {
      (n,getSimScore(xData(n),event))
    }).seq

    val (z,s) = if (rs.size > 0) rs.maxBy(_._2) else ("", 0.0)
    
    val res = if (s >= S) {
      val t = mTime(z)
      val eventD = xData(z)
      val sizeD = eventD.m("m").size
      val vs = merge(eventD.v,sizeD,event.v,event.m("m").size)
      val g = eventD.s
      val ms = eventD.m("m") ++ event.m("m")

      val ct = eventD.c
//        if (eventD.ct.getTime < event.ct.getTime) eventD.ct
//        else event.ct
      val ut =
        if (eventD.u.getTime < event.u.getTime) event.u
        else eventD.u
        
      val index = M1(z) // Update Index
      mIndex(t).remove(index, eventD.v.map(_._1).toArray)
      mIndex(t).insert(index, vs.map(_._1).toArray)
      
      val e = new Topic
      e.i = z
      e.m = Map("m"->ms)
      e.v = vs
      e.c = ct
      e.u = ut
      e.s = g
      mDB.insert(t,z,e) // Update Feature
      
      if (isNew == false){ // 需要删除老数据
        val ii = M1(event.i)
        val tt = mTime(event.i)
        mIndex(tt).remove(ii,event.v.map(_._1).toArray) // 删除索引
        val ee = new Topic
        ee.i = event.i
        ee.c = event.c
        ee.u = event.u
        ee.m = Map("m"->Array(z))
        mDB.insert(tt,ee.i,ee) // 覆盖老数据
        _log.info("Remove "+ee.i)
      } // 存在删除不成功的情况
      
      CACHE(z) = e // Update Cache
      
      z
    }
    else {
      if (isNew == false) { // 已经存在
    	""
      } 

      else { // New
        val i = M1(event.i)
        insert(event.i, i, event)
        CACHE(event.i) = event // Update Cache
        event.i
      }
    }
    
    res
  }
  
  
  def process(
		  	M1:HMap[String,Long],
		  	M2:HMap[Long,String],
		  	x: Document,
		  	S: Double
      ) = {
    
    val event = asTopic(x)
    
    // retrieve
    //_log.info("Retrieve ...")
    val ns = retrieve(event.v,minS).filter(M2.contains(_)).map(M2(_)).toArray
    if (ns.size > 10000){
      val line = event.v.map(vv => {
        vv._1+":"+vv._2
      }).reduce(_+" "+_)
      _log.info(line)
      val xline = event.v.map(vv => {
        vv._1+":"+mVU.get(vv._1)
      }).reduce(_+" "+_)
      _log.info(xline)
    }
    //_log.info("Retrieve Done")
    
    // Get Feature
    val xData = getFeature(ns)
    
    // Cosine
    val rs = ns.par.map(n => {
      (n,getSimScore(xData(n),event))
    }).seq
    
    //_log.info("Update ...")
    val (z,s) = if (rs.size > 0) rs.maxBy(_._2) else ("", 0.0)
    
    val res = if (s >= S){
      val t = mTime(z)
      val eventD = xData(z)
      val vs = eventD.v
      val g = eventD.s
      val ms = eventD.m("m") ++ event.m("m")
      
      val ct = 
        if (eventD.c.getTime < event.c.getTime) eventD.c
        else event.c
      val ut =
        if (eventD.u.getTime < event.u.getTime) event.u
        else eventD.u
      
      val e = new Topic
      e.i = z
      e.m = Map("m"->ms)
      e.v = vs
      e.c = ct
      e.u = ut
      e.s = g
      mDB.insert(t,z,e)
      
      CACHE(z) = e // Update Cache
      
      z
    }
    else {
      val i = M1(event.i)
      insert(event.i,i,event)
      
      CACHE(event.i) = event // Update Cache
      
      event.i
    }
    //_log.info("Update Done")
    
    res
    
  }
  

}
